package com.cly.common.config;


import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.PostConstruct;
import java.util.UUID;

@Configuration
@RequiredArgsConstructor
@Slf4j
@ConditionalOnClass(RabbitTemplate.class)
public class MqConfig {

    private final RabbitTemplate rabbitTemplate;

    /**
     * 配置生产者消息确认机制
     */
    @PostConstruct
    public void init() {
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                log.error("触发return callback,");
                log.info("exchange:{}", returnedMessage.getExchange());
                log.info("routingkey:{}", returnedMessage.getRoutingKey());
                log.info("message:{}", returnedMessage.getMessage());
                log.info("replyCode:{}", returnedMessage.getReplyCode());
                log.info("replyText:{}", returnedMessage.getReplyText());
            }
        });
    }

    /**
     * 配置消费者重试机制结束后，如果消息还没处理掉，就发送失败信息到 error.queue
     * @return
     */
    @Bean
    public MessageRecoverer republishMessageRecoverer(){
        return new RepublishMessageRecoverer( rabbitTemplate,
                "error.direct", "error");
    }

    /**
     * 设置ConfirmCallback回调函数
     * @return
     */
    @Bean
    public CorrelationData initCorrelationData(){
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                // mq自己的错误，基本不会触发
                log.error("spring amqp 处理确认结果异常", ex);
            }
            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                if(result.isAck()){
                    log.info("发送消息成功，收到回调 ACK！");
                }else{
                     log.error("发送消息失败，收到回调 NACK！:{}", result.getReason());
                }
            }
        });
        return  cd;
    }

    /**
     * 消息序列化方式
     * @return
     */
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}
